package com.xy.job.admin.service

import com.xy.job.admin.api.request.JobInstanceQueryParam
import com.xy.job.admin.api.response.JobInstancePageItemDTO
import com.xy.job.admin.common.buildEntityFromMap
import com.xy.job.admin.core.ExecutorClient
import com.xy.job.admin.domain.JobInfo
import com.xy.job.admin.domain.JobInstance
import com.xy.job.admin.event.JobInstanceCreateEvent
import com.xy.job.admin.repository.JobInfoRepository
import com.xy.job.admin.repository.JobInstanceRepository
import com.xy.job.core.api.request.JobKillParam
import com.xy.job.core.enums.JobStatus
import com.xy.job.core.exception.BizException
import org.slf4j.LoggerFactory
import org.springframework.context.ApplicationEventPublisher
import org.springframework.data.domain.Page
import org.springframework.data.domain.PageRequest
import org.springframework.data.domain.Pageable
import org.springframework.data.domain.Sort
import org.springframework.stereotype.Service
import org.springframework.transaction.annotation.Transactional
import java.time.LocalDateTime


@Service
class JobInstanceService (
    var jobInfoRepository: JobInfoRepository,
    var jobInstanceRepository: JobInstanceRepository,
    var eventPublisher: ApplicationEventPublisher,
){

    private final val log = LoggerFactory.getLogger(JobInstanceService::class.java)

    fun list(param: JobInstanceQueryParam): Page<JobInstancePageItemDTO> {
        param.triggerTimeBegin = param.triggerTimeBegin ?: LocalDateTime.of(1000, 1, 1, 0, 0, 0, 0)
        param.triggerTimeEnd = param.triggerTimeEnd ?: LocalDateTime.of(9999, 1, 1, 0, 0, 0, 0)
        val pageable: Pageable = PageRequest.of(
            param.pageNo - 1, param.pageSize, Sort.by(JobInfo::id.name).descending()
        )
        val page = jobInstanceRepository.findSimplePage(param, pageable)
        return page.map { buildEntityFromMap<JobInstancePageItemDTO>(it) }
    }

    fun detail(jobInstanceId: Long): JobInstance {
        return jobInstanceRepository.findById(jobInstanceId).orElseThrow {
            throw BizException("jobInstance not exist, id=[$jobInstanceId]")
        }
    }

    fun getErrorMessage(jobInstanceId: Long): String? {
        val jobInstance = jobInstanceRepository.findById(jobInstanceId).orElseThrow {
            throw BizException("jobInstance not exist, id=[$jobInstanceId]")
        }
        return jobInstance.message
    }

    fun kill(jobInstanceId: Long) {
        val jobInstance = jobInstanceRepository.findById(jobInstanceId).orElseThrow {
            throw BizException("jobInstance not exist, jobInstanceId=[$jobInstanceId]")
        }
        if (jobInstance.jobStatus != JobStatus.PROCESSING) {
            throw BizException("jobStatus is not 'PROCESSING', jobInstanceId=[$jobInstanceId]")
        }
        val executorClient = ExecutorClient(
            host = jobInstance.executorHost!!,
            port = jobInstance.executorPort!!,
        )
        val responseWrapper = executorClient.kill(JobKillParam(jobInstanceId = jobInstanceId))
        if (responseWrapper.success && responseWrapper.data == true) {
            log.info("kill job success. jobInstanceId={}", jobInstanceId)
        } else {
            throw BizException("kill job failed. jobInstanceId=$jobInstanceId, message=${responseWrapper.message}")
        }
    }

    @Transactional
    fun rerun(jobInstanceId: Long) {
        val originJobInstance = jobInstanceRepository.findById(jobInstanceId).orElseThrow {
            throw BizException("jobInstance not exist, jobInstanceId=[$jobInstanceId]")
        }
        val jobInfo = jobInfoRepository.findById(originJobInstance.jobId!!).orElseThrow {
            throw BizException("job not exist, jobId=[$originJobInstance.jobId]")
        }
        val existProcessingInstanceCount = jobInstanceRepository.countByJobIdAndJobStatus(
            jobId = jobInfo.id!!,
            jobStatus = JobStatus.PROCESSING,
        )
        if (existProcessingInstanceCount >= jobInfo.maxConcurrentNum!!) {
            throw BizException("当前任务实例已达到最大并发数，无法继续创建新的任务实例")
        }
        val allowJobStatus = listOf(JobStatus.FAILED, JobStatus.SUCCESS)
        if (originJobInstance.jobStatus !in allowJobStatus) {
            throw BizException("retry is only available when jobStatus in ${allowJobStatus}, jobInstanceId=[$jobInstanceId]")
        }
        val newJobInstance = jobInfo.createInstance().apply {
            this.executeParams = originJobInstance.executeParams
            this.dataTime = originJobInstance.dataTime
        }
        jobInstanceRepository.save(newJobInstance)
        eventPublisher.publishEvent(JobInstanceCreateEvent(newJobInstance))
    }
}